package com.raven.play.stream.test.javadsl.tutorial;

import akka.Done;
import akka.NotUsed;
import akka.japi.Pair;
import akka.stream.FlowShape;
import akka.stream.SinkShape;
import akka.stream.SourceShape;
import akka.stream.UniformFanInShape;
import akka.stream.javadsl.*;
import com.raven.play.stream.test.BaseTest;
import org.junit.jupiter.api.Test;

import java.util.concurrent.CompletionStage;

public class $10OpenGraph extends BaseTest {

    @Test
    public void test1(){
        Source<Integer, NotUsed> firstSource = Source.range(1, 10);
        Source<Integer, NotUsed> secondSource = Source.range(51, 100);

        Source<Object, NotUsed> sourceGraph = Source.fromGraph(GraphDSL.create(builder -> {

            SourceShape<Integer> firstSourceShape = builder.add(firstSource);
            SourceShape<Integer> secondSourceShape = builder.add(secondSource);

            UniformFanInShape<Object, Object> concatShape = builder.add(Concat.create(2));

            builder.from(firstSourceShape).toFanIn(concatShape);
            builder.from(secondSourceShape).toFanIn(concatShape);

            return SourceShape.of(concatShape.out());
        }));
        sourceGraph.runForeach(it -> System.out.println("it = " + it), system);
    }

    @Test
    public void test2(){
        Source<Integer, NotUsed> input = Source.range(1, 10);
        Flow<Integer, Integer, NotUsed> incrementer = Flow.of(Integer.class).map(it -> it + 1);
        Flow<Integer, Integer, NotUsed> multiplier = Flow.of(Integer.class).map(it -> it * 2);
        Sink<Pair<Integer, Integer>, CompletionStage<Done>> output = Sink.foreach(it -> log.info(it.toString()));

        Flow<Integer, Integer, NotUsed> flowGraph = Flow.fromGraph(GraphDSL.create(builder -> {

            FlowShape<Integer, Integer> incrementerShape = builder.add(incrementer);
            FlowShape<Integer, Integer> multiplierShape = builder.add(multiplier);

            builder.from(incrementerShape).via(multiplierShape);

            return FlowShape.of(incrementerShape.in(), multiplierShape.out());
        }));

        input.via(flowGraph).runForeach(it -> System.out.println("it = " + it), system);
    }

}
